package lab3;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import lab3.PageRankIter.IterMap;
import lab3.PageRankIter.IterReduce;

import org.apache.hadoop.mapreduce.Reducer;

public class PageRankViewer {
	public static class ViewMap extends Mapper<LongWritable,Text,DoubleWritable,Text>{	
		public void map(LongWritable key,Text value,Context context)
				throws  IOException, InterruptedException{
			String line = value.toString();
			// pagekey + \t +pr + \t + pr_link
			String [] tuple = line.split("\t");
			double pr_total = Double.parseDouble(tuple[1]);
			context.write(new DoubleWritable(pr_total) ,new Text(tuple[0]));
		}
	}
	public static class ViewReduce extends Reducer<DoubleWritable,Text,Text,NullWritable>{
		public void reduce (DoubleWritable key,Iterable<Text> values, Context context) 
				throws  IOException, InterruptedException{
			for(Text v:values) {
				String v_str = v.toString(); 
				String res = "(" + v_str + ","+String.format("%.10f", key.get())+")";
				context.write(new Text(res),NullWritable.get());
			}
		}
	}
	
	// 要求是从大到小排
	public static class MyDoubleComparator extends DoubleWritable.Comparator {
	    // @Override
	    public float compare(WritableComparator a,
	        WritableComparable<DoubleWritable> b) {
	      return -super.compare(a, b);
	    }

	    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
	      return -super.compare(b1, s1, l1, b2, s2, l2);
	    }
	  }
	
	public static void main(String[] args) throws Exception{
		// TODO Auto-generated method stub
		 Configuration conf = new Configuration();
		  Path path =new Path(args[1]);
		  FileSystem fileSystem = path.getFileSystem(conf);
	      if (fileSystem.exists(new Path(args[1]))) { 
	      	fileSystem.delete(new Path(args[1]),true); 
	      }
	      
		  Job job = new Job(conf, "PageRank View");
		  job.setJarByClass(PageRankViewer.class);
		  job.setMapperClass(ViewMap.class);
		  job.setReducerClass(ViewReduce.class);
		  // 重新定义比较函数
		  job.setSortComparatorClass(MyDoubleComparator.class);
		  job.setMapOutputKeyClass(DoubleWritable.class);  
	      job.setMapOutputValueClass(Text.class);  
	          
	      job.setOutputKeyClass(Text.class);  
	      job.setOutputValueClass(NullWritable.class); 
	      
		  FileInputFormat.addInputPath(job, new Path(args[0]));
		  FileOutputFormat.setOutputPath(job, new Path(args[1]));
		 job.waitForCompletion(true);
	}

}
